/*
 *   Copyright 1999-2016 Asiainfo Technologies(China),Inc.
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 *
 */

package com.asiainfo.redis.service.busidatawriter;

import redis.clients.util.SafeEncoder;

import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * @version v 1.0 on 2016/7/28 11:26
 * @auther william.xu
 */
public class BusiDataRedisWriter4ThreeMoreConditionsSaveConditionKeyValues extends BusiDataRedisWriter {

    private String conditionFieldName;

    public BusiDataRedisWriter4ThreeMoreConditionsSaveConditionKeyValues(Integer operFlag, String conditionFieldName){
        super(operFlag);
        this.conditionFieldName = conditionFieldName;
    }

    @Override
    public void doWrite(ResultSet rst) throws Exception {
        String busiTypeCode = busiTypeRepository.findOne(busiDataRule.getBusiTypeId()).getBusiCode();
        String keyPrefix = busiDataRule.getKeyPrefix();
        String oldConditionFieldValue = null;
        byte[] key = null;
        Set<String> list = new HashSet<>();
        while (rst.next()) {
            Object value = rst.getObject(conditionFieldName);
            String conditionFieldValue = (value==null?"":value.toString());
            if (oldConditionFieldValue == null){
                oldConditionFieldValue = conditionFieldValue;
                key = SafeEncoder.encode(busiTypeCode+":"+keyPrefix+":"+conditionFieldName+":"+oldConditionFieldValue);
            }
            if (conditionFieldValue != null && !conditionFieldValue.equals(oldConditionFieldValue)){
                setRedisKeyValue(key, list);
                //清空之前的list，重置oldValue和key的值
                list = new HashSet<>();
                oldConditionFieldValue = conditionFieldValue;
                key = SafeEncoder.encode(busiTypeCode+":"+keyPrefix+":"+conditionFieldName+":"+oldConditionFieldValue);
            }
            //把取到的unique数据加到list中
            list.add(rst.getObject(busiDataRule.getUniqueField()).toString());
        }
        //遍历结束，把最后一段数据写入redis
        if (key != null) {
            setRedisKeyValue(key, list);
        }
    }

    private void setRedisKeyValue(byte[] key, Set<String> list)
            throws Exception{
        List<byte[]> value = new ArrayList<>();
        for (String s : list){
            value.add(SafeEncoder.encode(s));
        }
        connection.del(key);
        if (operFlag == 0) {
            connection.sAdd(key, value.toArray(new byte[0][]));
        }
        publishMessage(key);
    }
}
